package com.baosight.xdata.zyl;

import com.baosight.xdata.redis.ConsumerInfo;
import com.baosight.xdata.redis.LettuceUtil;
import com.baosight.xdata.redis.StreamConsumerInfo;
import io.lettuce.core.*;
import io.lettuce.core.models.stream.PendingMessage;

import java.time.Duration;
import java.util.*;
import java.util.stream.Collectors;

public class TssRedisStreamClear {
    private int batch = 5;

    public static String STREAM_1 = "zyl-stream:1";
    public static String STREAM_2 = "zyl-stream:2";

    public static String GROUP_1 = "zyl-group1";
    public static String GROUP_2 = "zyl-group2";

    public static String CONSUMER_1 = "zyl-consumer1";
    public static String CONSUMER_2 = "zyl-consumer2";
    public static String CONSUMER_3 = "zyl-consumer3";
    public static String CONSUMER_4 = "zyl-consumer4";

    private void initConsumer() {
        Set<RedisURI> uris = new HashSet<>();
        uris.add(RedisURI.builder().withHost("10.25.10.102").withPort(6378).withTimeout(Duration.ofSeconds(30)).build());
        uris.add(RedisURI.builder().withHost("10.25.10.102").withPort(6377).withTimeout(Duration.ofSeconds(30)).build());
        uris.add(RedisURI.builder().withHost("10.25.10.102").withPort(6379).withTimeout(Duration.ofSeconds(30)).build());
        uris.add(RedisURI.builder().withHost("10.25.10.104").withPort(6378).withTimeout(Duration.ofSeconds(30)).build());
        uris.add(RedisURI.builder().withHost("10.25.10.104").withPort(6377).withTimeout(Duration.ofSeconds(30)).build());
        uris.add(RedisURI.builder().withHost("10.25.10.104").withPort(6379).withTimeout(Duration.ofSeconds(30)).build());
        LettuceUtil.init(uris);
    }

    public void consumerStreamViaGroup() {
        XReadArgs xReadArgs = new XReadArgs();
        xReadArgs.block(Duration.ofSeconds(5));
        xReadArgs.count(batch);
        List<StreamMessage<String, String>> list1;
        List<StreamMessage<String, String>> list2;
        while (true) {
            list1 = LettuceUtil.xread(xReadArgs, XReadArgs.StreamOffset.from(STREAM_1, "0-0"));
            handlerMessages(list1);
            LettuceUtil.xdel(STREAM_1, list1);

            list2 = LettuceUtil.xread(xReadArgs, XReadArgs.StreamOffset.from(STREAM_2, "0-0"));
            handlerMessages(list1);
            LettuceUtil.xdel(STREAM_2, list2);
        }
    }

    public void consumerPendingViaGroup(boolean enableClear) {
        XReadArgs xReadArgs = new XReadArgs();
        xReadArgs.block(Duration.ofSeconds(5));
        xReadArgs.count(batch);
        List<StreamConsumerInfo> streamConsumerInfoList = new ArrayList<>();
        streamConsumerInfoList.add(new StreamConsumerInfo(STREAM_1, GROUP_1, null));
        streamConsumerInfoList.add(new StreamConsumerInfo(STREAM_2, GROUP_2, null));
        for (StreamConsumerInfo consumer : streamConsumerInfoList) {
            System.out.println(String.format("========== %s ==========", consumer.toString()));
            consumerPendingViaGroup(consumer, enableClear);
        }
    }

    public List<ConsumerInfo> getConsumerInfo() {
        List<StreamConsumerInfo> streamConsumerInfoList = new ArrayList<>();
        streamConsumerInfoList.add(new StreamConsumerInfo(STREAM_1, GROUP_1, CONSUMER_1));
        streamConsumerInfoList.add(new StreamConsumerInfo(STREAM_1, GROUP_1, CONSUMER_3));
        streamConsumerInfoList.add(new StreamConsumerInfo(STREAM_2, GROUP_2, CONSUMER_2));
        streamConsumerInfoList.add(new StreamConsumerInfo(STREAM_2, GROUP_2, CONSUMER_4));
        List<ConsumerInfo> ConsumerInfoList = new ArrayList<>();
        for (StreamConsumerInfo streamConsumerInfo : streamConsumerInfoList) {
            List<Object> consumerInfo = LettuceUtil.getConsumerInfo(streamConsumerInfo.getStream(), streamConsumerInfo.getGroup());
            consumerInfo.stream().forEach(x -> ConsumerInfoList.add(ConsumerInfo.Builder((ArrayList<Object>) x)));
        }
        return ConsumerInfoList;
    }

    public void consumerPendingViaGroup(StreamConsumerInfo streamConsumerInfo, boolean enableClear) {
        XReadArgs xReadArgs = new XReadArgs();
        xReadArgs.block(Duration.ofSeconds(5));
        xReadArgs.count(batch);
        String lowerBound = "-";
        String upperBound = "+";
        List<ConsumerInfo> consumerInfo = getConsumerInfo(streamConsumerInfo);
        Range<String> range = Range.from(Range.Boundary.including(lowerBound), Range.Boundary.including(upperBound));
        for (ConsumerInfo entry : consumerInfo) {
            System.out.println(String.format("\n\nstream:[%s] group:[%s] consumer:[%s]", streamConsumerInfo.getStream(), streamConsumerInfo.getGroup(), streamConsumerInfo.getGroup()));
            if (entry.getPending() <= 0) {
                System.out.println("pending size [0]");
            } else {
                for (long i = entry.getPending(); i > 0; ) {
                    if(!lowerBound.equals("-")){
                        range = Range.from(Range.Boundary.excluding(lowerBound), Range.Boundary.including(upperBound));
                    }

                    List<PendingMessage> pendingMessageList = LettuceUtil.xpendingGroup(streamConsumerInfo.getStream(),
                            streamConsumerInfo.getGroup(), entry.getName(), range
                            , Limit.from(batch));
                    System.out.println(String.format("\n\nstream:[%s] group:[%s] consumer:[%s] pending size [%d]",
                            streamConsumerInfo.getStream(),
                            streamConsumerInfo.getGroup(),
                            streamConsumerInfo.getGroup(),
                            pendingMessageList.size()
                    ));
                    pendingMessageList.forEach(x -> System.out.println(x.toString()));
                    List<String> collect = pendingMessageList.stream().map(PendingMessage::getId).collect(Collectors.toList());
                    if(enableClear){
                        LettuceUtil.xackWithBatch(streamConsumerInfo.getStream(), streamConsumerInfo.getGroup(), collect);
                    }else{
                        collect.forEach(x -> System.out.println(x));
                        lowerBound = collect.get(collect.size() - 1 );
                    }
                    i = i - (pendingMessageList.size() > 0 ?  pendingMessageList.size() : batch);
                }
            }
        }
    }

    public List<ConsumerInfo> getConsumerInfo(StreamConsumerInfo streamConsumerInfo) {
        List<ConsumerInfo> ConsumerInfoList = new ArrayList<>();
        List<Object> consumerInfo = LettuceUtil.getConsumerInfo(streamConsumerInfo.getStream(), streamConsumerInfo.getGroup());
        for (Object entry : consumerInfo) {
            ArrayList<Object> list = (ArrayList<Object>) entry;
            ConsumerInfoList.add(ConsumerInfo.Builder(list));
        }
        return ConsumerInfoList;
    }


    private void handlerMessages(List<StreamMessage<String, String>> messageList) {
        if (!messageList.isEmpty()) {
            StringBuilder stringBuilder = new StringBuilder("=============");
            messageList.stream().forEach(x -> stringBuilder.append(x.toString()).append("\n"));
            System.out.println(stringBuilder.append("=============\n\n").toString());
        }
    }

    public static void main(String[] args) {
        TssRedisStreamClear tssRedisStreamConsumer = new TssRedisStreamClear();
        tssRedisStreamConsumer.initConsumer();
//        tssRedisStreamConsumer.consumerStreamViaGroup();
        tssRedisStreamConsumer.consumerPendingViaGroup(true);
    }


}
